package yz.mr.MapJoin;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
/**
 *  MapJoin:
 *      将数据关联的过程放在Map端进行处理
 *      那么Map函数是一行处理一次，那么如果做关联
 *      关联逻辑：按照读取一行时，需要另一部分数据中根据Key进行遍历循环
 *      所以需要两部分数据
 *
 * 加载两部分数据
 *
 */
public class MapJoin {
    public static void main(String[] args) throws Exception{

        //1.创建Job操作对象
        //2.创建配置类对象
        Configuration entries = new Configuration();
        Job job = Job.getInstance(entries);

        //3.配置job
        job.setJobName("MapJoin");
        job.setJarByClass(MapJoin.class);

        //4.设置Mapper以及输出类型
        job.setMapperClass(MapJoinMapper.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(MapStudent.class);

        //6.设置最终地输出数据类型
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(MapStudent.class);

        //添加学生的缓存路径
        job.addCacheFile(new Path("output/scoreCount/part-r-00000").toUri());

        //7.设置输入输出路径(这是本地的路径)  输入路径既可以指定路径也可以指定目录
        FileInputFormat.addInputPath(job,new Path("output/ReduceCount"));
        FileOutputFormat.setOutputPath(job,new Path("output/MapJoin"));

        //8.提交执行当前JOB
        job.waitForCompletion(true);

    }
}
